/**
 * @author zhounan 
 * @date 2024年6月19日
 * @version V1.0
 */
package org.zn.etl.job.context;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

/**
 * @author zhounan
 * @date 2024年6月19日
 * @version V1.0
 */
@Component
@Slf4j
public class NodeRunningContextManager implements ApplicationListener<ClusterJobEvent> {

	private Map<String, JobExecInstanceContext> runningContext;

	@Autowired(required = false)
	private ClusterContextManager clusterContextManager;

	private String self;

	public String getHostAddress() {
		return self;
	}

	@PostConstruct
	public void init() throws UnknownHostException {
		runningContext = new HashMap<>();
		self = InetAddress.getLocalHost().getHostAddress();

	}

	public void putJobExecIns(JobExecInstanceContext ins) throws Exception {

		String nodeManagerEnable = (String) ins.getContextMap().get(GlobalContextKey.CLUSTER_MAG_ENB);

		if (nodeManagerEnable != null && "false".equals(nodeManagerEnable.toLowerCase())) {
			return;
		}

		ins.setExecNode(self);

		String jobkey = StringUtils.isNotEmpty(ins.getClusterJobName()) ? ins.getClusterJobName()
				: ins.getJob().getJobName();

		if (clusterContextManager != null) {
			if (runningContext.containsKey(jobkey)) {
				clusterContextManager.updateJobExecIns(ins);
			} else {
				clusterContextManager.addJobExecIns(ins);
			}
		}

		runningContext.put(ins.getJob().getJobName(), ins);

	}

	public boolean renmoveJobExecIns(String name) {

		boolean r= runningContext.remove(name)!=null;
		if(clusterContextManager!=null) {
		r=	clusterContextManager.remove(name);			
		}
		
		return r;

	}

	public JobExecInstanceContext findNode(String name) {
		return runningContext.get(name);

	}

	public List<JobExecInstanceContext> getNodeList() {
		return new ArrayList<>(runningContext.values());

	}

	public List<ClusterJobExecInfo> getClusterList() {
		return clusterContextManager.getAll();

	}

	public ClusterJobExecInfo findCluster(String name) {

		ClusterJobExecInfo nji = clusterContextManager.find(name);

		return nji;

	}

	public void cancel(String jobName) {

		JobExecInstanceContext jeic = runningContext.get(jobName);
		jeic.setDoCmd(JobRunningCmd.cancel);

		renmoveJobExecIns(jobName);

	}

	public void killJob(String jobName) {

		JobExecInstanceContext jeic = runningContext.get(jobName);
		jeic.setDoCmd(JobRunningCmd.kill);
		jeic.getExecuteThread().interrupt();

		renmoveJobExecIns(jobName);

	}

	public boolean pause(String jobName) throws InterruptedException {

		JobExecInstanceContext jeic = runningContext.get(jobName);

		if (jeic != null) {
			jeic.setDoCmd(JobRunningCmd.pause);

			log.info("job:{}  , thread:{} do waiting", jobName, jeic.getExecuteThread().getId());

			jeic.getExecuteThread().wait();

			return true;
		}
		return false;

	}

	public boolean resume(String jobName) {

		JobExecInstanceContext jeic = runningContext.get(jobName);
		if (jeic != null) {
			jeic.setDoCmd(JobRunningCmd.resume);

			log.info("job:{}  , thread:{} do notify", jobName, jeic.getExecuteThread().getId());
			jeic.getExecuteThread().notify();

			return true;
		}

		return false;

	}

	@Override
	public void onApplicationEvent(ClusterJobEvent event) {
		// TODO Auto-generated method stub

		JobExecInstanceContext jei = (JobExecInstanceContext) event.getSource();
		log.debug("job:{}  , update ClusterJobEvent ", jei.getJob().getJobName());

		if(clusterContextManager!=null) {
			clusterContextManager.updateJobExecIns(jei);			
		}
		

	}

}
